<?php

namespace App\Console\Commands;

use App\Events\DeviceError;
use App\Events\DeviceStatusMessageProcess;
use App\Events\OrderStatusChange;
use Illuminate\Console\Command;

use Log;

use App\Support\Mqtt\MqttClient;
use React\EventLoop\Factory;

class MqttSubscribe extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'mqtt:subscribe';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '启动 Mqtt订阅服务';

    /**
     * Create a new command instance.
     *
     */
    public function __construct()
    {
        parent::__construct();
    }

    /**
     * Execute the console command.
     *
     * @return mixed
     */
    public function handle()
    {
        $server = config('mqtt.host');     // change if necessary
        $port = config('mqtt.port');                     // change if necessary
        $username = config('mqtt.username');                   // set your username
        $password = config('mqtt.password');                   // set your password                   // change if necessary
//        $username = 'admin';                   // set your username
//        $password = 'admin';                   // set your password
        $client_id = "webapp-subscribe"; // make sure this is unique for connecting to sever - you could use uniqid()

        $loop = Factory::create();

        $client = new MqttClient($server, $port);
        $client->setAuthentication($username,$password);
        $client->setDebug(true);
        $client->sendConnect($client_id, false, 30);  // set your client ID

        $loop->addPeriodicTimer(10, function () use (&$client, $client_id) {
            if (!$client->isConnected()) {
                $client->sendConnect($client_id, false, 30);  // set your client ID
                $this->setSubscribes($client);
            }
        });

        $loop->nextTick(function () use ($client) {
            $this->setSubscribes($client);
        });

        $loop->addPeriodicTimer(3, function () use (&$client) {
            $client->sendPing();
        });

        $loop->addPeriodicTimer(0, function () use (&$client) {
            $messages = $client->getPublishMessages(10000,false);
            // 处理消息
            $this->processMessages($messages);
        });

        $loop->run();
    }

    /**
     * 订阅Mqtt主题
     * @param $client
     */
    public function setSubscribes($client)
    {
        $client->sendSubscribe([
            'stop',
            'stop2',
            'start',
            'error',
            '$sys/dev/alarm',
            '$sys/client/connect',
            '$sys/client/disconnect',
        ], MqttClient::MQTT_QOS1);
    }

    /**
     * 处理订阅消息
     * @param $messages
     */
    public function processMessages($messages)
    {
        foreach ($messages as $item) {
            switch ($item['topic']) {
                case '$sys/client/disconnect':
                    call_user_func_array([$this, 'deviceDisConnect'], array($item['message']));
                    continue;
                case '$sys/client/connect':
                    call_user_func_array([$this, 'deviceConnect'], array($item['message']));
                    continue;
                case 'stop':
                    call_user_func_array([$this, 'orderStop'], array($item['message']));
                    continue;
                case 'stop2':
                    call_user_func_array([$this, 'orderStop2'], array($item['message']));
                    continue;
                case 'start':
                    call_user_func_array([$this, 'orderStart'], array($item['message']));
                    continue;
                case 'error':
                    call_user_func_array([$this, 'deviceError'], array($item['message']));
                    continue;
            }
        }
    }

    /**
     * 处理订单停止计费
     * @param $message
     */
    private function orderStop ($message)
    {
        Log::debug('order-stop', [$message]);
        $params = explode(';', $message);
        event(new OrderStatusChange($params[0], 'stop'));
    }

    /**
     * 处理订单停止计费2类型
     * @param $message
     */
    private function orderStop2($message)
    {
        Log::debug('order-stop-2', [$message]);
        $params = explode(';', $message);
        event(new OrderStatusChange($params[0], 'stop2'));
    }

    /**
     * 处理订单开始计费
     * @param $message
     */
    private function orderStart ($message)
    {
        Log::debug('order-start', [$message]);
        $params = explode(';', $message);
        event(new OrderStatusChange($params[0], 'start'));
    }

    /**
     * 处理设备上线
     * @param $message
     */
    private function deviceConnect ($message)
    {
        $params = json_decode($message, true);

        Log::debug('device-connect', [$params]);
        event(new DeviceStatusMessageProcess($params['client_id'], 'online'));
    }

    /**
     * 处理设备掉线
     * @param $message
     */
    private function deviceDisConnect ($message)
    {
        $params = json_decode($message, true);

        Log::debug('device-disconnect', [$params]);

        event(new DeviceStatusMessageProcess($params['client_id'], 'offline'));
    }

    /**
     * 处理设备掉线
     * @param $message
     */
    private function deviceError ($message)
    {
        $params = explode(';', $message);

        Log::debug('device-error', [$params]);

        event(new DeviceError($params[0], $params[1]));
    }
}
